package com.margo.project.consumer.kafka.client.controller;


import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/kafka/productor")
public class MargoKafkaProductorController {
	private Logger logger=LoggerFactory.getLogger(this.getClass());

	@Autowired
    private KafkaProducer<String, String> kafkaProducer;
	
	@RequestMapping("/send")
	public int send() {
		ProducerRecord<String, String> record=new ProducerRecord<String, String>("margo-kafka-topic",0,"name","margo");
		kafkaProducer.send(record,new Callback() {
			public void onCompletion(RecordMetadata metadata, Exception exception) {
				// TODO Auto-generated method stub
				logger.info("topic:"+metadata.topic());
				logger.info("offset:"+metadata.offset());
				logger.info("partition:"+metadata.partition());
			}
		});
		return 1;
	}
}
